package com.mjk.common.core.rmq.mq;


import com.mjk.common.tools.logger.BizLogger;

import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class PriorityBlockingQueue<E extends Runnable> implements BlockingQueue<E> {

	private final Map<Integer, Queue<E>> queueMap;

	private final List<Integer> prioritys;

	private final AtomicInteger count;

	private final ReentrantLock lock = new ReentrantLock();

	private final Condition empty = lock.newCondition();

	private final Condition notEmpty = lock.newCondition();

	private final int capacity;

	public PriorityBlockingQueue(List<Integer> prioritys, int capacity) {
		this.capacity = capacity;
		this.queueMap = new HashMap<>();
		if (prioritys == null) {
			this.prioritys = new ArrayList<>();
			int defaultPriority = 0;
			this.prioritys.add(defaultPriority);
		} else {
			this.prioritys = new ArrayList<>(prioritys);
		}
		for (Integer p : this.prioritys) {
			LinkedList<E> queue = new LinkedList<>();
			queueMap.put(p, queue);
		}
		if (prioritys != null) {
			Collections.sort(prioritys);
		}
		this.count = new AtomicInteger();
	}

	@Override
	public int size() {
		return this.count.get();
	}

	@Override
	public boolean isEmpty() {
		return this.count.get() == 0;
	}

	protected E get() {
		for (int p : this.prioritys) {
			Queue<E> queue = getQueue(p);
			E data = queue.poll();
			if (data != null) {
				return data;
			}
		}
		return null;
	}

	@Override
	public E poll(long timeout, TimeUnit unit) {
		lock.lock();
		try {
			E data = get();
			if (data != null) {
				this.count.decrementAndGet();
				notEmpty.signal();
				return data;
			} else {
				if(empty.await(timeout, unit)) {
                    return get();
                } else {
                    return null;
                }
			}
		} catch (Exception e) {
			Thread.currentThread().interrupt();
			return null;
		} finally {
			lock.unlock();
		}
	}

	@Override
	public E take() {
		lock.lock();
		try {
			E data;
			while ((data = get()) == null) {
				empty.await();
			}
			this.count.decrementAndGet();
			notEmpty.signal();
			return data;
		} catch (Exception e) {
			Thread.currentThread().interrupt();
			return take();
		} finally {
			lock.unlock();
		}
	}

	@Override
	public boolean offer(E e) {
		if (e == null) {
			throw new NullPointerException();
		}
		lock.lock();
		try {
			while (this.capacity <= size()) {
				notEmpty.await();
			}
			boolean success = getQueue(e).offer(e);
			if (success) {
				this.count.incrementAndGet();
				empty.signal();
				return true;
			} else {
				return offer(e);
			}
		} catch (Exception e1) {
			Thread.currentThread().interrupt();
			BizLogger.error(e1);
			return false;
		} finally {
			lock.unlock();
		}
	}

	@SuppressWarnings("unchecked")
	@Override
	public boolean remove(Object task) {
		if (!(task instanceof RunnablePriority)) {
			return true;
		}
		lock.lock();
		try {
			E e = (E) task;
			boolean success = getQueue(e).remove(e);
			if (success) {
				this.count.decrementAndGet();
				notEmpty.signal();
			}
			return success;
		} catch (Exception e1) {
			return false;
		} finally {
			lock.unlock();
		}
	}

	protected Queue<E> getQueue(int priority) {
		return this.queueMap.get(priority);
	}

	protected Queue<E> getQueue(E e) {
		if (!(e instanceof RunnablePriority)) {
			return getFristQueue();
		}
		return getQueue(((RunnablePriority) e).getPriority());
	}

	protected Queue<E> getFristQueue() {
		return getQueue(this.prioritys.get(0));
	}

	@Override
	public E element() {
		return null;
	}

	@Override
	public E peek() {
		return null;
	}

	@Override
	public E poll() {
		return null;
	}

	@Override
	public E remove() {
		return null;
	}

	@Override
	public boolean addAll(Collection<? extends E> c) {
		return false;
	}

	@Override
	public void clear() {
		// Do nothing
	}

	@Override
	public boolean containsAll(Collection<?> c) {
		return false;
	}

	@Override
	public Iterator<E> iterator() {
		return null;
	}

	@Override
	public boolean removeAll(Collection<?> c) {
		return false;
	}

	@Override
	public boolean retainAll(Collection<?> c) {
		return false;
	}

	@Override
	public Object[] toArray() {
		return new Object[0];
	}

	@Override
	public <T> T[] toArray(T[] a) {
		return a;
	}

	@Override
	public boolean add(E e) {
		return false;
	}

	@Override
	public boolean contains(Object o) {
		return false;
	}

	@Override
	public int drainTo(Collection<? super E> c) {
		return 0;
	}

	@Override
	public int drainTo(Collection<? super E> c, int maxElements) {
		return 0;
	}

	@Override
	public boolean offer(E e, long timeout, TimeUnit unit) {
		return false;
	}

	@Override
	public void put(E e) {
		// Do nothing
	}

	@Override
	public int remainingCapacity() {
		return 0;
	}

}
